package com.unitd.scheduler.helper;

import com.unitd.scheduler.job.JobContext;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.*;

/**
 * @desc 自定义的用于处理一致性的一致性哈希处理工具类
 * @filename ConsistencyHash.java
 * @copyright www.unitd.com
 * @author Hudan
 * @version 1.0
 * @date 2016/10/14
 */
public class ConsistencyHash {

	/** 定时任务的部署节点数 */
	private TreeMap<Long, String> nodes = new TreeMap<>();
	/** 设置虚拟节点数目,默认4节点 */
	private final int VIRTUAL_NUM = 4;

	/**
	 * @desc 刷新部署服务的节点
	 * @param shards 节点列表
	 */
	public void refresh(List<String> shards) {
		nodes.clear();
		for (int i = 0; i < shards.size(); i++) {
			String shardInfo = shards.get(i);
			for (int j = 0; j < VIRTUAL_NUM; j++) {
				nodes.put(hash(i + "-" + j), shardInfo);
			}
		}
	}

	/**
	 * @desc 根据hash因子获取分配的真实节点
	 * @param factor 节点对象的key
	 * @return 节点
	 */
	public String getAssignedRealNode(Object factor) {
		// 如果不存在部署的定时服务节点, 从定时器工厂类中获取
		if (nodes.size() <= 1) return JobContext.getContext().getNodeId();

		Long key = hash(factor.toString());
		SortedMap<Long, String> tailMap = nodes.tailMap(key);
		if (tailMap.isEmpty()) {
			key = nodes.firstKey();
		} else {
			key = tailMap.firstKey();
		}
		return nodes.get(key);
	}


	/**
	 * @desc 根据部署节点的key,对该节点key进行hash取值,获取最终的hash值
	 * @param key 节点key
	 * @return 节点对应的hash值
	 */
	private static Long hash(String key) {
		if (key == null) return 0L;
		ByteBuffer buf = ByteBuffer.wrap(key.getBytes());
		int seed = 0x1234ABCD;

		ByteOrder byteOrder = buf.order();
		buf.order(ByteOrder.LITTLE_ENDIAN);

		long m = 0xc6a4a7935bd1e995L;
		int r = 47;

		long h = seed ^ (buf.remaining() * m);

		long k;
		while (buf.remaining() >= 8) {
			k = buf.getLong();

			k *= m;
			k ^= k >>> r;
			k *= m;

			h ^= k;
			h *= m;
		}

		if (buf.remaining() > 0) {
			ByteBuffer finish = ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN);
			finish.put(buf).rewind();
			h ^= finish.getLong();
			h *= m;
		}

		h ^= h >>> r;
		h *= m;
		h ^= h >>> r;

		buf.order(byteOrder);
		return Math.abs(h);
	}

	public static void main(String[] args) {
		ConsistencyHash hash = new ConsistencyHash();
		hash.refresh(new ArrayList<>(Arrays.asList("aa", "bbbbbbbbbbb")));

		for (int i = 1; i < 11; i++) {
			System.out.println(hash.getAssignedRealNode(i));
		}
	}
}